package org.yonggan.cmcc.report.biz

import org.apache.spark.rdd.RDD
import org.yonggan.cmcc.cache.JedisPoolUtil

/**
  * 统计实时交易性数据
  */
object RealTimeReportTradeData {


  /**
    * 实时统计每分钟的充值笔数和充值金额
    *     //   组转成为List : (订单总量 ,成功订单量， 金额 ，支付时长) , 省份编码 ,day h m）
    */
  def reportTradeDataByMin(rdd: RDD[(List[Double],String,String,String,String)]): Unit = {

    // 计算结果
    val result: RDD[((String, String), List[Double])] = rdd.map(f => {
      // 成功订单量
      val day = f._3
      val hoursMin = f._4 + "-" + f._5
      //      成功订单量， 金额
//      val lst = f._1.tail.take(2)
      ((day, hoursMin), List(f._1(1),f._1(2)))
    }).reduceByKey((lst1, lst2) => lst1.zip(lst2).map(m => m._1 + m._2))

    result.foreach(println)
    result.foreachPartition(itor => {
      val jedis = JedisPoolUtil.getJedis
      itor.foreach(f => {
        jedis.hincrBy("M-"+f._1._1,"S-"+f._1._2,f._2(0).toLong)//每分钟总量
        jedis.hincrByFloat("M-"+f._1._1,"M-"+f._1._2,f._2(1))//每分钟总金额
      })
      jedis.close()

    })
  }
}
